<?php
/**
 * Created by PhpStorm.
 * User: lee
 * Date: 2018/11/13
 * Time: 2:43 PM
 */

namespace Biz\service;

use Biz\pockets\constInterface\EventTag;
use function Clue\StreamFilter\fun;
use Doraemon\amqp\Queue;
use Doraemon\pockets\amqp\Amqp;
use Doraemon\tools\Log;
use PhpAmqpLib\Message\AMQPMessage;
/**
 * @deprecated
 */
class Event extends Amqp implements EventTag
{


    const _CONFIG_FILE_ = 'mq_event';

    const _CONFIG_SELECT_ = "event";

    const EVENT_SUB = [
        'order' => \Biz\module\Classes\OrderEvent::class,
        //'shop'
    ];

    //发送事件最大重试次数
    const PUSH_MAX_RETRIES = 3;

    //启用信号处理时捕捉的信号转换后的结果,0正常运行,1结束当前进程
    public static $signalResult = 0;
    public static $inConsume = 0;


    protected static $consumeHandle;
    protected static $maxReConnectNum = 5;




    public static function push($eventTag, $data = [], $eventId = 0,$msgId='')
    {

        $subClass = self::getClass($eventTag);

        if ($subClass === false) {
            return false;
        }

        if (!is_array($data)) {
            $data = $data();
        }

        if (empty($eventId)) {
            $eventId = self::genUniqId();
        }

        if(empty($msgId)){
            if(isset($data['uniq_idx'])){
                $data['event_tag'] = $eventTag;
                $msgId = self::genUniqIdx($data);
            }
        }

        //取得rabbitmq的配置
        $exchange = $subClass::EVENT_PUSH_EXCHANGE;

        $objSub = new $subClass;

        //push之前的回调
        if (method_exists($objSub, 'beforePush')) {
            $data = $objSub->beforePush($eventTag, $data, $eventId);
        }

        //构造消息
        $buildMsg = function ($eventTag, $data, $eventId) use ($msgId){
            $data = [
                'header' => [
                    'msg_id' => $msgId,
                    'event_tag' => $eventTag,
                    'event_id' => $eventId,
                    'addtime' => date('Y-m-d H:i:s')
                ],
                'body' => $data
            ];

            $msg = self::buildAMQPMessage(json_encode($data, JSON_UNESCAPED_UNICODE));

            return $msg;
        };

        //初始化Rabbmit链接
        $channel = self::_getChannel(false, $eventTag);

        for ($i = 0; $i < self::PUSH_MAX_RETRIES; $i++) {
            try {

                $channel->exchange_declare($exchange, 'topic', false, true, false);

                //需要对发出的消息进行到达确认
                $channel->tx_select();

                $msg = $buildMsg($eventTag, $data, $eventId);


                //发送消息
                $channel->basic_publish($msg, $exchange, $eventTag);

                //到达确认需要的方法调用
                $channel->tx_commit();

                //push之后的回调
                if (method_exists($objSub, 'afterPush')) {
                    $objSub->afterPush($eventTag, $data, $eventId);
                }

                Log::event_push_info('push事件成功', [
                    'func_args' => func_get_args(),
                    'msg' => $msg,
                    'exchange' => $exchange,
                ]);

                return true;

            } catch (\PhpAmqpLib\Exception\AMQPChannelClosedException $e) {
                Log::event_push_error('push事件出错，消息回滚', [
                    'func_args' => func_get_args(),
                    'e_file' => $e->getFile(),
                    'e_code' => $e->getCode(),
                    'e_msg' => $e->getMessage()
                ]);


                //事件push任务处理异常
                Queue::insertQueue("queue/Wechatnotice/toTag", [
                    "totag" => \Biz\pockets\constInterface\BizWechatTagConfig::TAG_BIZ_RD_SYSTEM,
                    "content" => vsprintf('任务Event::push异常,info=%s,e_msg=%s,e_code=%s,e_line=%s,e_file=%s', [
                        json_encode([
                            'event_tag' => $eventTag,
                            'event_id' => $eventId,
                            'data' => $data
                        ], JSON_UNESCAPED_UNICODE),
                        $e->getMessage(),
                        $e->getCode(),
                        $e->getLine(),
                        $e->getFile()
                    ])
                ], Queue::QUEUE_BIZ);

                $channel->tx_rollback();
            }
        }


        return false;

    }

    /**
     * $eventTag  支持通配符
     */

    public static function consume($eventTag, $func, $clientId, $timeout = 5)
    {


        $eventTags = $eventTag;

        if (!is_array($eventTag)) {
            $eventTags = [$eventTag];
        }

        $subClass = self::getClass($eventTags[0]);

        if ($subClass === false) {
            return false;
        }

        $exchange = $subClass::EVENT_PUSH_EXCHANGE;

        $queueName = self::genQueueName($eventTags, $clientId);

        //信号处理
        self::signalCallback($queueName);

        self::setErrorHandler($queueName);


        $callback = function ($message) use ($func, $timeout, $queueName) {
            try {
                if (!empty(Event::$signalResult)) {
                    self::reject($message);
                    self::signalExit($queueName, false);
                }
                Event::$inConsume = 1;

                $data = json_decode($message->body, true);

                if (empty($data)) {
                    self::ack($message);
                }

                $func($message, $data['header']['event_tag'], $queueName);

                Event::ack($message);

                Event::$inConsume = 0;

                if (!empty(Event::$signalResult)) {
                    self::signalExit($queueName, false);
                }

            } catch (\Exception $e) {
                self::reject($message);
            }
        };

        $forceConnect = false;
        $reConnectNum = 0;

        while (true) {
            try {

                self::$consumeHandle = self::_getChannel($forceConnect, 'consume');


                self::$consumeHandle->exchange_declare($exchange, 'topic', false, true, false);

                self::$consumeHandle->queue_declare($queueName, false, true, false, false);

                foreach ($eventTags as $eventTag) {
                    self::$consumeHandle->queue_bind($queueName, $exchange, $eventTag);
                }

                $forceConnect = false;
                $reConnectNum = 0;

                self::$consumeHandle->basic_consume($queueName, '', false, false, false, false, $callback);

                while (count(self::$consumeHandle->callbacks)) {
                    self::$consumeHandle->wait();
                }
            } catch (\PhpAmqpLib\Exception\AMQPChannelClosedException $e) {

                $forceConnect = true;
                $reConnectNum++;

                Log::event_consumeconnect_error('consume链接异常，第' . $reConnectNum . '次 重试连接', [
                    'queue_name' => $queueName,
                    'e_file' => $e->getFile(),
                    'e_code' => $e->getCode(),
                    'e_msg' => $e->getMessage()
                ]);

                if ($reConnectNum > self::$maxReConnectNum) {

                    //事件消费任务出现异常
                    Queue::insertQueue("queue/Wechatnotice/toTag", [
                        "totag" => \Biz\pockets\constInterface\BizWechatTagConfig::TAG_BIZ_RD_SYSTEM,
                        "content" => vsprintf('任务Event::consume异常,event_tags=%s,e_msg=%s,e_code=%s,e_line=%s,e_file=%s', [
                            implode('-', $eventTags),
                            $e->getMessage(),
                            $e->getCode(),
                            $e->getLine(),
                            $e->getFile()
                        ])
                    ], Queue::QUEUE_BIZ);


                    self::signalExit($queueName, self::$inConsume);

                } else {

                    sleep(1);

                    continue;

                }

            } catch (\Exception $e) {
                //事件消费任务出现异常
                Queue::insertQueue("queue/Wechatnotice/toTag", [
                    "totag" => \Biz\pockets\constInterface\BizWechatTagConfig::TAG_BIZ_RD_SYSTEM,
                    "content" => vsprintf('任务Event::consume出现 消费 异常,queue_name=%s,e_msg=%s,e_code=%s,e_line=%s,e_file=%s', [
                        $queueName,
                        $e->getMessage(),
                        $e->getCode(),
                        $e->getLine(),
                        $e->getFile()
                    ])
                ], Queue::QUEUE_BIZ);

                self::signalExit($queueName, self::$inConsume);
            }
        }

    }


    public static function ack($message)
    {
        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
    }

    public static function reject($message)
    {
        $message->delivery_info['channel']->basic_reject($message->delivery_info['delivery_tag'], true);
    }

    public static function genUniqId()
    {
        $ms = microtime();
        $ms = substr($ms, 2, 5);
        $eventId = date('YmdHis') . $ms . rand(10000, 99999);
        return (string)$eventId;
    }
    public static function genUniqIdx(&$data=[])
    {
        $map = array_map(function ($v) use(&$data){
            $v = isset($data[$v])?$data[$v]:'';
            return $v;
        },$data['uniq_idx']);

        $uniqBizId = sha1(json_encode($map,JSON_UNESCAPED_UNICODE));

        unset($data['uniq_idx']);
        unset($data['event_tag']);
        return $uniqBizId;
    }

    protected static function getClass($eventTag)
    {
        $prefix = substr($eventTag, 0, strpos($eventTag, '.'));
        if (empty(self::EVENT_SUB[$prefix])) {
            return false;
        }
        $subClass = self::EVENT_SUB[$prefix];
        return $subClass;
    }

    protected static function genQueueName($eventTags, $clientId = '')
    {
        $prefix = 'queue.ha.event.biz.order_';
        $hostName = gethostname();
        $suffix = 'from.biz_event.consume';
        $clientId = !empty($clientId) ? $clientId . '_' : '';

        return $prefix . $hostName . '_' . $clientId . $suffix;
    }

    protected static function signalCallback($queueName = '')
    {
        declare(ticks=1);

        $signalCallback = function ($signo) use ($queueName) {
            Event::$signalResult = 1;

            if (empty(Event::$inConsume)) {
                self::signalExit($queueName, false);
            }
        };

        pcntl_signal(SIGTERM, $signalCallback);
    }

    protected static function closeChannel()
    {
        try {

            self::$consumeHandle->close();
        } catch (\Exception $e) {

        }
    }

    protected static function signalExit($queueName = '', $inConsume = false)
    {
        Log::event_consume_error('系统被 kill', [
            'queue_name' => $queueName,
            'in_consume' => $inConsume ? ' 正在消费中,被 kill' : '没有在消费中的被kill'
        ]);

        Queue::insertQueue("queue/Wechatnotice/toTag", [
            "totag" => \Biz\pockets\constInterface\BizWechatTagConfig::TAG_BIZ_RD_SYSTEM,
            "content" => vsprintf('事件消费的队列,queue_name=%s,info=%s由于系统维护或其他原因，被KILL，请知晓', [
                $queueName,
                json_encode([
                    'in_consume' => $inConsume ? ' 正在消费中,被 kill' : '没有在消费中的被kill'
                ], JSON_UNESCAPED_UNICODE)
            ])
        ], Queue::QUEUE_BIZ);

        self::closeChannel();

        //关闭channel后等待1秒
        sleep(1);

        posix_kill(posix_getpid(), SIGKILL);
    }

    protected static function setErrorHandler($queueName)
    {
        global $argv;
        set_error_handler(function ($errno, $errstr, $errfile, $errline) use ($queueName, $argv) {
            switch ($errno) {
                case E_USER_ERROR:
                    Queue::insertQueue("queue/Wechatnotice/toTag", [
                        "totag" => \Biz\pockets\constInterface\BizWechatTagConfig::TAG_BIZ_RD_SYSTEM,
                        "content" => vsprintf('事件消费的队列,queue_name=%s,info=%s由于系统出现Error级别错误而被KILL，请相关负责人处理，并重启该任务=%s', [
                            'queue_name' => $queueName,
                            json_encode([
                                'error_info' => [
                                    'errorno' => $errno,
                                    'errorstr' => $errstr,
                                    'errorfile' => $errfile,
                                    'errorline' => $errline
                                ]
                            ], JSON_UNESCAPED_UNICODE),
                            isset($argv[1]) ? $argv[1] : '无'
                        ])
                    ], Queue::QUEUE_BIZ);
                    break;
            }

        });
    }


}